package com.example.rocketmq.demos;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.HashSet;
import java.util.Set;

@Slf4j
public class PullConsumer {

    public static void main(String[] args) throws MQClientException {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("PullConsumer");
        consumer.setNamesrvAddr("192.168.101.141:9876");
        HashSet<String> topics = new HashSet<>();
        topics.add("test-topic");
        consumer.setRegisterTopics(topics);
        consumer.start();
        consumer.getRegisterTopics().forEach(topic->{
            try {
                Set<MessageQueue> queues = consumer.fetchSubscribeMessageQueues(topic);
                for (MessageQueue messageQueue : queues) {
                    long offset = consumer.getOffsetStore().readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY);
                    PullResult result = consumer.pull(messageQueue, "*", offset, 32);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

}
